package polimi.disttest;

import java.net.ConnectException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Random;

import polimi.reds.Filter;
import polimi.reds.Message;
import polimi.reds.TCPDispatchingService;
import polimi.reds.optimizer.OptimizerStrategyOptions;

public class Component extends Thread {

	private TCPDispatchingService d;
	private OptimizerStrategyOptions o;
	private ObjectRandomizer<Integer> randomizer;
	private Random random;
	private boolean terminated=false;
	
	public Component(OptimizerStrategyOptions o) {
		d=new TCPDispatchingService("reds-tcp:" + o.getOption(LocalManager.OPTION_IP) + ":" + o.getOption(LocalManager.OPTION_PORT));
		this.o = o;
		this.random = new Random(this.hashCode() + o.getIntegerOption(LocalManager.OPTION_SEED));
		
		List<Integer> events = new ArrayList<Integer>();
		for (int i=0; i<o.getIntegerOption(LocalManager.OPTION_MESSAGE_CLASSES); i++) {
			events.add(i);
		}
		Collections.shuffle(events, random);
		randomizer = new ObjectRandomizer<Integer>(events, random);
	}

	public synchronized void startTermination() {
		terminated=true;
	}
	
	public void run() {
		try {
			d.open();
		} catch (ConnectException e) {
			System.err.println("Unable to connect the components: " + e.getMessage());
			System.exit(0);
		}
		
		ArrayList<GenericFilter> filters = new ArrayList<GenericFilter>();
		
		while (filters.size() < o.getIntegerOption(LocalManager.OPTION_NUM_COMPONENT_SUBSCRIPTIONS)) {
			GenericFilter f = createSubscription();
			if (!filters.contains(f)) {
				filters.add(f);
				d.subscribe(f);
			}
		}
		
		do {
			try {
				
				sleep((int) (1+Math.abs(random.nextGaussian()*o.getIntegerOption(LocalManager.OPTION_DELAY_MESSAGES_STDEV)+o.getIntegerOption(LocalManager.OPTION_DELAY_MESSAGES_MEAN))));
			} catch (InterruptedException e) {
			}
			
			while (d.hasMoreMessages()) {
				d.getNextMessage();
			}
			
			d.publish(createMessage());
			
		} while (!terminated);
		d.close();
	}
	
	private synchronized GenericFilter createSubscription() {
		GenericFilter filter = new GenericFilter();
			
		for (Integer i=0; i<o.getIntegerOption(LocalManager.OPTION_CLASSES_PER_SUBSCRIPTION); i++) {
			
			if (o.getOption(LocalManager.OPTION_SUBSCRIPTIONS_CLASSES_DISTRIBUTION).equals("zipf")) {
				filter = filter.union(new GenericFilter(new GenericMessage(randomizer.nextObjectZipf(filter.getPrototypeMessagesID()))));
			} else if (o.getOption(LocalManager.OPTION_SUBSCRIPTIONS_CLASSES_DISTRIBUTION).equals("uniform")) {
				filter = filter.union(new GenericFilter(new GenericMessage(randomizer.nextObjectUniform(filter.getPrototypeMessagesID()))));
			} else {
				System.err.println("Unknown message classes distribution: " + o.getOption(LocalManager.OPTION_MESSAGES_CLASSES_DISTRIBUTION));
				System.exit(0);
			}
		}
		return filter;	
	}
	
	private GenericMessage createMessage() {
		if (o.getOption(LocalManager.OPTION_MESSAGES_CLASSES_DISTRIBUTION).equals("zipf")) {
			return new GenericMessage(randomizer.nextObjectZipf());
		} else if (o.getOption(LocalManager.OPTION_MESSAGES_CLASSES_DISTRIBUTION).equals("uniform")) {
			return new GenericMessage(randomizer.nextObjectUniform());
		} else {
			System.err.println("Unknown message classes distribution: " + o.getOption(LocalManager.OPTION_MESSAGES_CLASSES_DISTRIBUTION));
			System.exit(0);
			return null;
		}
	}

}
